package drds.server.route.sequence;

import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import drds.server.Server;
import drds.server.config.ErrorCode;

public class SequnceProcessor {
	private static final Logger LOGGER = LoggerFactory.getLogger(SequnceProcessor.class);
	private LinkedBlockingQueue<SequenceSql> sequenceSqlQueue = new LinkedBlockingQueue<SequenceSql>();
	private volatile boolean running = true;

	public SequnceProcessor() {
		new ExecuteThread().start();
	}

	public void addSequenceSql(SequenceSql sequenceSql) {
		sequenceSqlQueue.add(sequenceSql);
	}

	private void executeSequenceSql(SequenceSql sequenceSql) {
		try {

			Sequencer sequencer = new Sequencer(Server.getInstance().getConfig().getDRDS().getSequnceHandlerType());

			String sql = sequencer.getExecuteSql(sequenceSql.sql);

			sequenceSql.session.getSessionContext().routeAndExecute(sql, sequenceSql.type, sequenceSql.schema);
		} catch (Exception e) {
			LOGGER.error("SequenceProcessor.executeSeq(SesionSQLPair)", e);
			sequenceSql.session.getSessionContext().writeErrorMessage(ErrorCode.ER_YES, " sequnce err." + e);
			return;
		}
	}

	public void shutdown() {
		running = false;
	}

	class ExecuteThread extends Thread {

		public ExecuteThread() {
			// 设置为后台线程,防止throw RuntimeExecption进程仍然存在的问题
			setDaemon(true);
		}

		public void run() {
			while (running) {
				try {
					SequenceSql sequenceSql = sequenceSqlQueue.poll(100, TimeUnit.MILLISECONDS);
					if (sequenceSql != null) {
						executeSequenceSql(sequenceSql);
					}
				} catch (Exception e) {
					LOGGER.error("SequenceProcessor$ExecutorThread", e);
				}
			}
		}
	}
}
